package com.atguigu.cn.dataStream.sink

import java.util

import com.atguigu.cn.dataStream.api.SensorReading
import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
import org.apache.http.HttpHost
import org.elasticsearch.client.Requests

/**
  * @author: yangShen
  * @Description:
  * @Date: 2020/4/15 14:36 
  */
class EsSinkTest {
  def main(args: Array[String]): Unit = {

    //1.environment
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)

    // 2.source
    val inputStream = environment.readTextFile("D:\\my\\myGit\\mayun\\miaohui8023\\my-flink\\flink-tutorial\\src\\main\\resources\\sensor.txt")

    // 3.transform 操作，但没有标记返回类型
    val dataStream = inputStream.map(data => {
      val dataArray = data.split(",")
      SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
    })

    val httpHosts = new util.ArrayList[HttpHost]()
    httpHosts.add(new HttpHost("localhost",9200))

    //创建一个esSink的builder
    val esSinkBuilder = new ElasticsearchSink.Builder[SensorReading](httpHosts,new ElasticsearchSinkFunction[SensorReading] {
      override def process(t: SensorReading, runtimeContext: RuntimeContext, requestIndexer: RequestIndexer): Unit = {
        println("save data" + t)
        //包装成一个Map或者JsonObject
        val json = new util.HashMap[String, String]()
        json.put("sensor_id", t.id)
        json.put("temperature", t.temperature.toString)
        json.put("timestamp", t.timestamp.toString)

        //创建index request 准备发送数据
        val indexRequest = Requests.indexRequest()
          .index("sensor")
          .`type`("readingdata")  //  elasticsearch6之后的版本不需要这个了
          .source(json)

        //利用requestIndexer发送请求，写入数据
        requestIndexer.add(indexRequest)

        println("data save")

      }
    })

    //sink
    dataStream.addSink(esSinkBuilder.build())

    environment.execute("es sink test")

    //结果在浏览器中确认
    //查看所有的索引：localhost:9200/_cat/indices?v
    //查看sensor索引的内容：localhost:9200/sensor/_search?pretty
  }
}
